package com.gzhryc.common.kafka;

import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

import com.gzhryc.common.logging.Logger;

public class KafkaConsumerFactory {
	
	private static final Logger log = Logger.getLogger(KafkaConsumerFactory.class);

	private static KafkaConsumerFactory consumerFactory = null ;

    private Properties properties;

    private KafkaConsumerFactory(){}

    public static KafkaConsumerFactory self(){
        if ( consumerFactory == null ) {
            consumerFactory = new KafkaConsumerFactory() ;
            return consumerFactory ;
        }
        return consumerFactory ;
    }
    
    /*  用于存放kafka的消费者对象  */
    ConcurrentHashMap<String, JFKafkaConsumer> kafkaMap = new ConcurrentHashMap<>() ;
    
    //  初始化ConsumerFactory
    public KafkaConsumerFactory initial(Properties properties){
        this.properties = properties;
        return this ;
    }

    public KafkaConsumerFactory addConsumer(JFKafkaConsumer consumer){
    	JFKafkaConsumer oldConsumer = kafkaMap.put(getKey(consumer.getTopic(),consumer.getGroupId()), consumer);
    	//避免重复添加
    	if(oldConsumer != null) {oldConsumer.close();}
    	consumer.start(properties);
    	log.info("添加消息接受者，接收主题：" + consumer.getTopic() + "，组ID：" + consumer.getGroupId());
        return this;
    }

    public JFKafkaConsumer getConsumer(String topic,String groupId) {
    	return kafkaMap.get(getKey(topic,groupId));
    }
    
    private String getKey(String topic,String groupId) {
    	return topic+"_"+groupId;
    }
}
